
#include "ae.h"
#include "thread.h"
#include "session.h"
#include "mylog.h"

#include <assert.h>
#include <errno.h>
#include <stdarg.h>
#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>

static long long sThreadId = 0;

static void onFileReadEvent(struct aeEventLoop * eventLoop, int fd, void *clientData, int mask) {
	mylog(LOG_INFO, "ONFILEREADEVETN!!!");
	struct session* session = (struct session* )clientData;
	if (!session) {
		return;
	}

	int len;
	len = readSession(session);
	mylog(LOG_DEBUG, "on file read event, session id[%d], len[%d]", session->id, len);
	if (len <= 0) {
		if (session->callBackHandle) {
			if (session->callBackHandle->onDisConnectionCallBack) {
				session->callBackHandle->onDisConnectionCallBack(session);
			}
		}

		destroySession(session);
	} else {
		int ret;
		if (session->callBackHandle) {
			if (session->callBackHandle->onDataCallBack) {
				ret = session->callBackHandle->onDataCallBack(session);
			}
		}

		mylog(LOG_INFO, "session id[%d], onDataCallBack ret[%d]", session->id, ret);
		if (ret == SHUTDOWN_HANDLE) {
			shutdownSession(session);
		} else if (ret == CONTINUE_HANDLE) {
			if (getReadableBytesBuffer(session->readBuffer) <= 0) {
				retrieveAllBuffer(session->readBuffer);
			}
		} else if (ret == CLOSE_HANDLE) {
			destroySession(session);
		}
	}
}

void handleFileReadEvent(struct session* session, struct thread* t) {
	mylog(LOG_INFO, "in handle File ReadEvent!!!");
	if (!session || !t ) {
		mylog(LOG_WARNING, "onMsgqueueEvent, session or thread is null!!!");
		return;
	}

	int fd = session->cfd;
	if (fd < 0) {
		mylog(LOG_WARNING, "onMsgqueueEvent some error, session id[%d], thread id[%d]", session->id, t->threadId);
		destroySession(session);
		return;
	}

	session->t = t;
	int ret = aeCreateFileEvent(t->threadEventLoop, fd, AE_READABLE, onFileReadEvent, session);
	if (ret == AE_ERR){
		mylog(LOG_WARNING, "onMsgqueueEvent some error, create File Event error!!! session id[%d], thread id[%d]", session->id, t->threadId);
		destroySession(session);
		return;
	}

	if (session->callBackHandle) {
		if (session->callBackHandle->onConnectionCallBack) {
			session->callBackHandle->onConnectionCallBack(session);
		}
	}
}

typedef struct AsyncSendContent {
	struct session* session;
	char* sendBuf;
	int sendBufLen;
} AsyncSendContent;

void handleSendEvent(struct AsyncSendContent* content, struct thread* t) {
	mylog(LOG_INFO, "ONSENDEVENT!!!");
	if (!content) {
		return;
	}

	struct session* session = content->session;
	char* sendBuf = content->sendBuf;
	int sendBufLen = content->sendBufLen;
	writeSession(session, sendBuf, sendBufLen);
	if (sendBuf) {
		zfree(sendBuf);
	}
	zfree(content);
}

void handleStopThread(struct thread* t) {
	if (!t) {
		return;
	}

	t->threadState =STOP_THREADSTATE;
}

void onMsgqueueEvent(void* arg1, void* arg2) {
	struct msgQueueEvent* event = (struct msgQueueEvent*)arg1;
	struct thread* t = (struct thread* )arg2;
	if (!event || !t) {
		return;
	}

	mylog(LOG_INFO, "ON MSGQUEUEeVENT!!!!!");
	if (event->eventType == FILEREADEVENT_TYPE) {
		handleFileReadEvent((struct session*)event->privateData, t);
	} else if (event->eventType == SENDEVENT_TYPE) {
		handleSendEvent((struct AsyncSendContent*)event->privateData, t);
	} else if (event->eventType == STOPEVENT_TYPE) {
		handleStopThread(t);
	}

	zfree(event);
}

void* eventLoopThread(void* args) {
	struct thread* t = (thread*) args;
	if (!t) {
		return NULL;
	}

	aeEventLoop* el = t->threadEventLoop;
	if (!el) {
		return NULL;
	}

	t->threadState = START_THREADSTATE;
	mylog(LOG_INFO, "eventloop thread[%d] start!!!", t->threadId);
	aeMain(t->threadEventLoop);
	mylog(LOG_INFO, "eventloop thread[%d] end!!!", t->threadId);
	return NULL;
}

int threadCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
	//mylog(LOG_INFO, "thread Cron!!!");
	struct thread* t = (struct thread*) clientData;
	if (!t) {
		return AE_NOMORE;
	}

	if (t->threadState == STOP_THREADSTATE) {
		mylog(LOG_INFO, "thread[%d] corn!!!", t->threadId);
		if (t->threadEventLoop) {
			aeStop(t->threadEventLoop);
			t->threadState = STOPING_THREADSTATE;
			return AE_NOMORE;
		}
	}
	return 100; /* every 100ms */
}

void stopThread(struct thread* t) {
	if (!t) {
		return;
	}

	mylog(LOG_INFO, "stop thread[%d]", t->threadId);
	t->threadState = STOP_THREADSTATE;
}

#define MSGQUEUESIZE	64
struct thread* initThread() {
	struct thread* t = NULL;
	if ( ! (t = (struct thread* )zcalloc(sizeof(struct thread)) ) ) {
		mylog(LOG_WARNING, "zcalloc thread error!!!");
		return NULL;
	}

	t->threadEventLoop = aeCreateEventLoop();
	if (! t->threadEventLoop) {
		mylog(LOG_WARNING, "in initThread create event loop error!!!");
		zfree(t);
		return NULL;
	}

	t->threadEventQueue = (struct eventMsgqueue*) initMsgqueue(t->threadEventLoop, MSGQUEUESIZE, onMsgqueueEvent, t);
	if (!t->threadEventQueue) {
		mylog(LOG_WARNING,  "in initThread init msg queue error!!!");
		aeDeleteEventLoop(t->threadEventLoop);
		zfree(t);
		return NULL;
	}

	t->threadId = ++sThreadId;
	int ret = pthread_create( &t->threadPid, NULL, eventLoopThread, (void* )t );
	if (ret == 0) {
		mylog(LOG_INFO, "create thread[%d] success!!!", t->threadId);
	} else {
		mylog(LOG_WARNING, "in initThread create thread[%d] error!!!\n", t->threadId);
		destroyMsgqueue(t->threadEventQueue);
		aeDeleteEventLoop(t->threadEventLoop);
		zfree(t);
		return NULL;
	}

	aeCreateTimeEvent(t->threadEventLoop,
										1,
										threadCron,
										t,
										NULL);
	t->threadState = INIT_THREADSTATE;
	return t;
}

void destroyThread(struct thread* t) {
	if (!t) {
		mylog(LOG_WARNING, "destroy thread error, thread t is null!!!\n");
		return;
	}

	mylog(LOG_INFO, "destroy thread[%d]ing!!!", t->threadId);

	pthread_t tid = t->threadPid;
	pthread_join(tid, NULL);

	mylog(LOG_INFO, "thread[%d] exit!!!", t->threadId);
	destroyMsgqueue(t->threadEventQueue);
	aeDeleteEventLoop(t->threadEventLoop);
	zfree(t);
	return;
}

void sendMsgInThread(struct session* session, char* sendBuf, int sendBufLen) {
	if (!session) {
		return;
	}

	struct thread* t = session->t;

	struct eventMsgqueue* theadEventQueue = t->threadEventQueue;
	if (! theadEventQueue) {
		mylog(LOG_WARNING, "in sendMsgInThread, threadEventQueue is null!!!");
		return;
	}

	struct msgQueueEvent* msgQueueEvent = (struct msgQueueEvent* )zcalloc(sizeof(struct msgQueueEvent));
	if (!msgQueueEvent) {
		return;
	}

	struct AsyncSendContent* sendContent = (struct AsyncSendContent*)zcalloc(sizeof(struct AsyncSendContent));
	if (!sendContent) {
		mylog(LOG_WARNING, "in despathWorkThread, sendcontent is null!!!");
		return;
	}
	sendContent->sendBuf = sendBuf;
	sendContent->sendBufLen = sendBufLen;
	sendContent->session = session;

	msgQueueEvent->eventType = SENDEVENT_TYPE;
	msgQueueEvent->privateData = sendContent;
	pushMsgqueue(theadEventQueue, msgQueueEvent);
}

void pushMsgQueueEvent(struct thread* t, struct session* session) {
	if (!t || !session) {
		return;
	}

	struct eventMsgqueue* threadEventQueue = t->threadEventQueue;
	if (! threadEventQueue) {
		mylog(LOG_WARNING, "in pushMsgQueueEvent, threadEventQueue is null!!!");
		return;
	}

	struct msgQueueEvent* msgQueueEvent = (struct msgQueueEvent* )zcalloc(sizeof(struct msgQueueEvent));
	if (!msgQueueEvent) {
		return;
	}

	msgQueueEvent->eventType = FILEREADEVENT_TYPE;
	msgQueueEvent->privateData = session;
	pushMsgqueue(threadEventQueue, msgQueueEvent);
}

void stopThreadEvent(struct thread* t) {
	if (!t) {
		return;
	}

	struct eventMsgqueue* threadEventQueue = t->threadEventQueue;
	if (! threadEventQueue) {
		mylog(LOG_WARNING, "in pushMsgQueueEvent, threadEventQueue is null!!!");
		return;
	}

	struct msgQueueEvent* msgQueueEvent = (struct msgQueueEvent* )zcalloc(sizeof(struct msgQueueEvent));
	if (!msgQueueEvent) {
		return;
	}

	msgQueueEvent->eventType = STOPEVENT_TYPE;
	msgQueueEvent->privateData = NULL;
	pushMsgqueue(threadEventQueue, msgQueueEvent);
}
